using System;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Threading;
using RxHeat.Common;
using RxHeat.Tracing;


namespace RxHeat
{
    /// <summary>
    ///     Wraps a source observable, caches and shares the observable subscription. Any new subscription attempts
    ///     within a specified <see cref="TimeSpan" /> would be funneled to an internal shared hot Observable.
    ///     <b>Disposing of this object would result in canceling/clearing the internal cached stream.</b>
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public sealed class CachedObservable<T> : IDisposable, IObservable<T>
    {
        private readonly TimeSpan _cachedObservableLifeSpan;
        private readonly IObservable<T> _instanceOfThrottledObservable;
        private readonly ReaderWriterLockSlim _rwLock = new ReaderWriterLockSlim(LockRecursionPolicy.NoRecursion);
        private readonly IScheduler _scheduler;
        private readonly SerialDisposable _serialDisposable = new SerialDisposable();
        private readonly IObservable<T> _sourceSubscription;
        private readonly CompositeDisposable _throttlingDisposable = new CompositeDisposable();
        private readonly string _tracerKey;
        private NetClassProxy.Tuple<IObservable<T>, DateTime> _cashedSubscription;

        /// <summary>
        ///     Initializes a new instance of the <see cref="CachedObservable{T}" /> class.
        /// </summary>
        /// <param name="wrappedObservable">The wrapped observable.</param>
        /// <param name="cachedObservableLifeSpan">The cached observable life span.</param>
        public CachedObservable(Func<IObservable<T>> wrappedObservable, TimeSpan cachedObservableLifeSpan)
            : this(wrappedObservable, cachedObservableLifeSpan, null)
        {
        }

        /// <summary>
        ///     Initializes a new instance of the <see cref="CachedObservable{T}" /> class.
        /// </summary>
        /// <param name="wrappedObservable">The wrapped observable.</param>
        /// <param name="cachedObservableLifeSpan">The cached observable life span.</param>
        /// <param name="scheduler">The scheduler.</param>
        public CachedObservable(Func<IObservable<T>> wrappedObservable, TimeSpan cachedObservableLifeSpan,
                                IScheduler scheduler)
            : this(Observable.Defer(wrappedObservable), scheduler, cachedObservableLifeSpan)
        {
        }

        /// <summary>
        ///     Prevents a default instance of the <see cref="CachedObservable{T}" /> class from being created.
        /// </summary>
        /// <param name="wrappedObservable">The wrapped observable.</param>
        /// <param name="scheduler">The scheduler.</param>
        /// <param name="cachedObservableLifeSpan">The cache life span.</param>
        private CachedObservable(IObservable<T> wrappedObservable, IScheduler scheduler,
                                 TimeSpan cachedObservableLifeSpan)
        {
            _tracerKey = GetType().DescribeType() + "." + GetHashCode();
            _sourceSubscription = wrappedObservable;
            _scheduler = scheduler;
            _serialDisposable.TrackDisposable(_throttlingDisposable);
            _cachedObservableLifeSpan = cachedObservableLifeSpan;
            _instanceOfThrottledObservable = GetObservable(_cachedObservableLifeSpan)
                .Publish()
                .RefCount();
        }

        /// <summary>
        ///     Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        public void Dispose()
        {
            if (!_throttlingDisposable.IsDisposed)
                _throttlingDisposable.Dispose();
        }

        /// <summary>
        ///     Subscribes the specified observer.
        /// </summary>
        /// <param name="observer">The observer.</param>
        /// <returns></returns>
        public IDisposable Subscribe(IObserver<T> observer)
        {
            return _instanceOfThrottledObservable.Subscribe(observer);
        }

        private IObservable<T> GetObservable(TimeSpan maximumAliveTimeSpan)
        {
            _rwLock.EnterUpgradeableReadLock();
            try
            {
                if (_cashedSubscription != null && _cashedSubscription.Item1 != null &&
                    (DateTime.Now - _cashedSubscription.Item2) < maximumAliveTimeSpan)
                    return _cashedSubscription.Item1;
                else
                {
                    _rwLock.EnterWriteLock();
                    try
                    {
                        _cashedSubscription = new NetClassProxy.Tuple<IObservable<T>, DateTime>(
                            Observable.Create<T>(
                                observer =>
                                (_scheduler == null ? _sourceSubscription : _sourceSubscription.SubscribeOn(_scheduler))
                                    .Subscribe(_tracerKey, observer)
                                    .TrackDisposable(_throttlingDisposable))
                                      .ToReplayObservable(_serialDisposable)
                            , DateTime.Now);
                        Tracing.ObservableExtensions.TraceDisposable(null, _tracerKey, "New Cashed Subscription");
                    }
                    finally
                    {
                        _rwLock.ExitWriteLock();
                    }
                    return _cashedSubscription.Item1;
                }
            }
            finally
            {
                _rwLock.ExitUpgradeableReadLock();
            }
        }
    }
}